package chapter5.nio;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by gzhd on 2017/11/27.
 */
public class NioServer {
    private static Selector mSelector;
    private static ExecutorService tp = Executors.newCachedThreadPool();
    public static Map<Socket, Long> time_stat = new HashMap<>();
    static  int port=8000;
    private static final int TIMEOUT = 3000;

    public static void main(String[] args) throws Exception {
        startServer();
    }

    private static void startServer() throws Exception {
        mSelector = SelectorProvider.provider().openSelector();
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        InetSocketAddress isa = new InetSocketAddress("127.0.0.1", port);
        ssc.socket().bind(isa);

        SelectionKey acceptKey = ssc.register(mSelector, SelectionKey.OP_ACCEPT);


        for (; ; ) {
            if (mSelector.select(TIMEOUT) == 0) {
                System.out.println("没有收到任何响应··再等");
                continue;
            }

            Set readyKeys = mSelector.selectedKeys();
            Iterator ite = readyKeys.iterator();
            long e = 0;
            while (ite.hasNext()) {
                SelectionKey sk = (SelectionKey) ite.next();
                ite.remove();
//                Socket tmpSocket = ((SocketChannel) sk.channel()).socket();
//                Socket tmpSocket = ((SocketChannel) sk.channel()).socket();
//                处理不同的逻辑
                if (sk.isAcceptable()) {
//                   处理连接
                    doAccept(sk);
                } else if (sk.isValid() && sk.isReadable()) {
                    if (!time_stat.containsKey(((SocketChannel) sk.channel()).socket())) {
                        time_stat.put(((SocketChannel) sk.channel()).socket(), System.currentTimeMillis());
                    }
//                    处理读
                    doRead(sk);
                } else if (sk.isValid() && sk.isWritable()) {
//                    处理写
                    doWrite(sk);
                    e = System.currentTimeMillis();
                    long b = time_stat.remove(((SocketChannel) sk.channel()).socket());
                    System.out.println("spend:" + (e - b) + "ms");
                }

            }

        }
    }

    private static void doWrite(SelectionKey sk) {
        SocketChannel channel = (SocketChannel) sk.channel();
        EchoClient echoClient = (EchoClient) sk.attachment();
        LinkedList<ByteBuffer> outq = echoClient.getOutputQueue();

        ByteBuffer bb = outq.getLast();
        try {
            int len = channel.write(bb);
            if (len == -1) {
                return;
            }
            if (bb.remaining() == 0) {
                outq.removeLast();
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
        if (outq.size() == 0) {
            sk.interestOps(SelectionKey.OP_READ);
        }

    }

    private static void doRead(SelectionKey sk) {
        SocketChannel channel = (SocketChannel) sk.channel();
        ByteBuffer bb = ByteBuffer.allocate(8192);
        int len;
        try {
            len = channel.read(bb);
            if (len < 0) {
//                disconnect(sk);
                return;
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
        bb.flip();
        tp.execute(new HandleMsg(sk, bb));
    }

    private static void doAccept(SelectionKey sk) {
        ServerSocketChannel server = (ServerSocketChannel) sk.channel();
        SocketChannel clientChannel;
        try {
            clientChannel = server.accept();
            clientChannel.configureBlocking(false);
            SelectionKey key = clientChannel.register(mSelector, SelectionKey.OP_READ);
            EchoClient echoClient = new EchoClient();
            key.attach(echoClient);
            InetAddress clientAddress = clientChannel.socket().getInetAddress();
            System.out.println("Accepted Connection from" + clientAddress.getHostAddress() + ".");

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }


    }


    static class HandleMsg implements Runnable {
        SelectionKey sk;
        ByteBuffer bb;

        @Override
        public void run() {
            EchoClient echoClient = (EchoClient) sk.attachment();
            echoClient.enqueue(bb);
            sk.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
            mSelector.wakeup();
        }

        public HandleMsg(SelectionKey sk, ByteBuffer bb) {
            this.sk = sk;
            this.bb = bb;
        }
    }


}
